package com.shujia.ycsb

import org.apache.spark.internal.Logging
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
import org.apache.spark.sql.expressions.Window
import org.slf4j.Logger

object DwsAppVioWfjltjMaxsb extends Logging {
  val logger: Logger = log

  def main(args: Array[String]): Unit = {

    var ds: String = ""
    // 从args里提取分区日期参数
    if (args.isEmpty) {
      logger.error("请指定一个分区日期！")
      return
    }
    ds = args.head

    // 构建SparkSession
    val spark: SparkSession = SparkSession
      .builder()
      .appName("DwsAppVioWfjltjMaxsb")
      .enableHiveSupport()
      .getOrCreate()

    // 导入隐式转换已经函数包
    import spark.implicits._
    import org.apache.spark.sql.functions._

    val resultDF: DataFrame = spark
      .table("dwd.dwd_base_vio_surveil")
      .where($"ds" === ds and $"sbbh".isNotNull and $"sbbh" =!= "")
      .select($"sbbh", date_format($"wfsj", "yyyy-MM-dd") as "wfsj_day")
      .groupBy($"sbbh", $"wfsj_day")
      // 计算每个设备每天的记录的违法数
      .agg(count("*") as "sbwfs_day")
      // 计算每个数据的位置
      .withColumn("rn", row_number() over Window.partitionBy($"sbbh").orderBy($"sbwfs_day"))
      // 分别计算q1和q3的位置
      .withColumn("q1_pos", ceil((count("*") over Window.partitionBy($"sbbh")) / 4))
      .withColumn("q3_pos", floor((count("*") over Window.partitionBy($"sbbh")) * 3 / 4))
      // 基于q1、q3位置提取对应位置上的值
      .withColumn("q1", sum(when($"rn" === $"q1_pos", $"sbwfs_day").otherwise(expr("0"))) over Window.partitionBy($"sbbh"))
      .withColumn("q3", sum(when($"rn" === $"q3_pos", $"sbwfs_day").otherwise(expr("0"))) over Window.partitionBy($"sbbh"))
      // 计算 上边界
      .withColumn("zdgjz", when($"q1" < $"q3", $"q3" + ($"q3" - $"q1") * 1.5).otherwise(expr("0")))
      // 过滤
      .where($"sbwfs_day" > $"zdgjz")
      // 整理数据
      .select(
        $"sbbh"
        , $"sbwfs_day"
        , $"zdgjz"
        , $"wfsj_day"
      )

    // 增加分区
    spark.sql(s"alter table dws.dws_app_vio_wfjltj_maxsb add if not exists partition(ds='$ds')")
    // 保存数据
    resultDF
      .write
      .mode(SaveMode.Overwrite)
      .format("csv")
      .option("sep", "^")
      .save(s"/daas/motl/dws/dws_app_vio_wfjltj_maxsb/ds=$ds")

  }

  /**
   * 提交命令：
   * spark-submit --master yarn-client --conf spark.sql.shuffle.partitions=2 --class com.shujia.ycsb.DwsAppVioWfjltjMaxsb car-1.0.jar 2023-06-14
   */
}
